package com.bridgeintelligent.tag.engine.mvc.service.impl;

import com.bridgeintelligent.tag.engine.api.*;
import com.bridgeintelligent.tag.engine.api.rsp.DimAnalysisRsp;
import com.bridgeintelligent.tag.engine.core.cache.CacheConfig;
import com.bridgeintelligent.tag.engine.core.cache.CacheUtils;
import com.bridgeintelligent.tag.engine.es.base.ESQueryUtils;
import com.bridgeintelligent.tag.engine.es.model.CustomersAnalysisModel;
import com.bridgeintelligent.tag.engine.es.model.GroupItem;
import com.bridgeintelligent.tag.engine.es.query.EsQueryBuilder;
import com.bridgeintelligent.tag.engine.mvc.service.ICustomersAnalysisService;
import com.bridgeintelligent.tag.engine.mvc.service.IEsDateService;
import com.bridgeintelligent.tag.engine.mvc.service.IGroupService;
import com.bridgeintelligent.tag.engine.mvc.service.IMonitorService;
import com.bridgeintelligent.tag.engine.parse.BaseParse;
import com.bridgeintelligent.tag.utils.FieldHandler;
import com.wayneleo.quickstart.framework.ExceptionCode;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.histogram.HistogramAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.range.Range;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @description
 * @author: grp 2020-11-05 13:47
 **/
@Service
@Slf4j
@AllArgsConstructor
public class CustomersAnalysisServiceImpl implements ICustomersAnalysisService, IMonitorService {

    final private IGroupService groupService;

    final private IEsDateService dateService;

    @Override
    public DimAnalysisRsp.R getSingleFieldValue(CustomersAnalysisModel model) throws IOException {
        DimInfo cacheInfo = CacheUtils.getCacheInfo(CacheConfig.DIM_INFO, model.getField());
        if(cacheInfo==null){
            throw new EngineException(new ExceptionCode(999,"未查询到标签"));
        }
        if(StringUtils.equals(cacheInfo.getClassify(),"04")){
            throw new EngineException(new ExceptionCode(999,"标识标签不支持"));
        }
        // 聚合策略
        if(StringUtils.equals(cacheInfo.getClassify(),"03")){
            return getNumberResult(model, cacheInfo);
        }
        if(StringUtils.equals(cacheInfo.getClassify(),"05")){
            return getDateResult(model, cacheInfo);
        }
        return getStringResult(model, cacheInfo);
    }

    @Override
    public List<DimAnalysisRsp.R> getFieldValues(CustomersAnalysisModel model) throws IOException {
        List<String> fields = fetchDateField(model.getFields());
        List<DimAnalysisRsp.R> rs = new ArrayList<>();
        for(String field:fields){
            model.setField(field);
            if(model.getItemsMap()!=null){
                // 设置对应分组
                model.setItems(model.getItemsMap().get(field));
            }else{
                model.setItems(null);
            }
            try{
                rs.add(getSingleFieldValue(model));
            }catch (Exception e){
                log.error(String.format("客群洞察 异常%s",field),e);
            }
        }
        return rs;
    }

    private List<String> fetchDateField(List<String> fieldsSet) {
        return fieldsSet.stream().distinct().filter(s -> !EngineConfig.CUST_NO.equals(s))
                .collect(Collectors.toList());
    }

    private DimAnalysisRsp.R getNumberResult(CustomersAnalysisModel model, DimInfo cacheInfo) throws IOException {
        // 筛选条件
        QueryBuilder queryBuilder = EsQueryBuilder.buildFilter(model.getFilterGroup());
        Map<String, String> relation = dateService.symbolRelation(model.getIndexId());
        String column = FieldHandler.addSuffixField(model.getField(),model.getFieldsDataDate(),relation);
        if(CollectionUtils.isEmpty(model.getItems())){
            return getNumberResult(column, queryBuilder, cacheInfo,model.getIndexId());
        }
        return getNumberResult(column, queryBuilder, cacheInfo, model.getItems(),model.getIndexId());
    }

    /**
     * 默认分组
     * @param column
     * @param queryBuilder
     * @param cacheInfo
     * @return
     * @throws IOException
     */
    private DimAnalysisRsp.R getNumberResult(String column, QueryBuilder queryBuilder, DimInfo cacheInfo,String indexId) throws IOException {
        Double interval = groupService.getNumberInterval(column, queryBuilder,indexId);
        if(interval==null){
            return DimAnalysisRsp.R.of().setInfo(Collections.emptyList()).setDimName(cacheInfo.getDimName()).setClassify(cacheInfo.getClassify()).setDimId(cacheInfo.getDimId());
        }
        SearchRequest request = getNumberHistogramRequest(queryBuilder, column, interval,indexId);
        return getNumberResult(request, interval, cacheInfo);
    }

    private DimAnalysisRsp.R getNumberResult(SearchRequest request, Double interval, DimInfo cacheInfo) throws IOException {
        // 处理结果
        SearchResponse response=
                ESQueryUtils.getClient().search(request, RequestOptions.DEFAULT);
        Histogram histogram = response.getAggregations().get("count");
        List<DimAnalysisRsp.Info> data = new ArrayList<>();
        for (Histogram.Bucket entry : histogram.getBuckets()) {
            long key = ((Double) entry.getKey()).longValue();
            long high = key + interval.longValue();
            data.add(DimAnalysisRsp.Info.of().setTitle(key+"~"+high).setCount(String.valueOf(entry.getDocCount())));
        }
        return DimAnalysisRsp.R.of().setInfo(data).setDimName(cacheInfo.getDimName()).setClassify(cacheInfo.getClassify()).setDimId(cacheInfo.getDimId());
    }

    private SearchRequest getNumberHistogramRequest(QueryBuilder queryBuilder,String dimIdColumn, Double interval,String indexId) {

        IndexNameInfo indexNameInfo = CacheUtils.getCacheInfo(CacheConfig.CONFIGURATION_INFO, indexId);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(queryBuilder).trackTotalHits(true).size(0);
        HistogramAggregationBuilder builder =
                AggregationBuilders.histogram("count").field(dimIdColumn).interval(interval).order(BucketOrder.key(true)).minDocCount(1);
        searchSourceBuilder.aggregation(builder);
        return new SearchRequest(indexNameInfo.getDataDate()).source(searchSourceBuilder);
    }

    /**
     * 自定义
     * @param column
     * @param queryBuilder
     * @param cacheInfo
     * @param items
     * @return
     */
    private DimAnalysisRsp.R getNumberResult(String column, QueryBuilder queryBuilder, DimInfo cacheInfo,
                                           List<GroupItem> items,String indexId) throws IOException {
        SearchRequest request = getRequest(column, queryBuilder, cacheInfo.getClassify(), items,indexId);
        return getResult(ESQueryUtils.getClient().search(request, RequestOptions.DEFAULT), cacheInfo);
    }

    private SearchRequest getRequest(String dimIdColumn, QueryBuilder queryBuilder, String type,
                                     List<GroupItem> items,String indexId) {

        IndexNameInfo indexNameInfo = CacheUtils.getCacheInfo(CacheConfig.CONFIGURATION_INFO, indexId);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(queryBuilder).trackTotalHits(true).size(0);
        if(StringUtils.equals("03", type)){
            searchSourceBuilder.aggregation(getNumberAggregation(items, dimIdColumn));
        }
        if(StringUtils.equals("05", type)){
            searchSourceBuilder.aggregation(getDateAggregation(items, dimIdColumn));
        }
        return new SearchRequest(indexNameInfo.getDataDate()).source(searchSourceBuilder);
    }

    private DimAnalysisRsp.R getResult(SearchResponse response, DimInfo cacheInfo){
        Range terms = response.getAggregations().get("count");
        List<DimAnalysisRsp.Info> data = new ArrayList<>();
        for (Range.Bucket entry : terms.getBuckets()) {
            data.add(DimAnalysisRsp.Info.of().setTitle(entry.getKeyAsString()).setCount(String.valueOf(entry.getDocCount())));
        }
        sort(cacheInfo.getClassify(), data);
        return DimAnalysisRsp.R.of().setInfo(data).setDimName(cacheInfo.getDimName()).setClassify(cacheInfo.getClassify()).setDimId(cacheInfo.getDimId());
    }

    private DimAnalysisRsp.R getDateResult(CustomersAnalysisModel model, DimInfo cacheInfo) throws IOException {
        // 筛选条件
        QueryBuilder queryBuilder = EsQueryBuilder.buildFilter(model.getFilterGroup());
        Map<String, String> relation = dateService.symbolRelation(model.getIndexId());
        String column = FieldHandler.addSuffixField(model.getField(),model.getFieldsDataDate(),relation);
        if(CollectionUtils.isEmpty(model.getItems())){
            return getDateResult(column, queryBuilder, cacheInfo,model.getIndexId());
        }
        return getDateResult(column, queryBuilder, cacheInfo, model.getItems(),model.getIndexId());
    }

    private DimAnalysisRsp.R getDateResult(String column, QueryBuilder queryBuilder, DimInfo cacheInfo,String indexId) throws IOException {
        DateHistogramInterval dateInterval = groupService.getDateInterval(column, queryBuilder, indexId);
        if(dateInterval==null){
            return DimAnalysisRsp.R.of().setInfo(Collections.emptyList()).setDimName(cacheInfo.getDimName()).setClassify(cacheInfo.getClassify()).setDimId(cacheInfo.getDimId());
        }
        SearchRequest request = getDateHistogramRequest(column, queryBuilder, dateInterval, indexId);
        return getDateResult(ESQueryUtils.getClient().search(request, RequestOptions.DEFAULT), dateInterval, cacheInfo);
    }

    private DimAnalysisRsp.R getDateResult(SearchResponse response, DateHistogramInterval interval, DimInfo cacheInfo) throws IOException {
        Histogram terms = response.getAggregations().get("count");
        List<DimAnalysisRsp.Info> data = new ArrayList<>();
        for (Histogram.Bucket entry : terms.getBuckets()) {
            data.add(DimAnalysisRsp.Info.of().setTitle(getTitle(entry.getKeyAsString(), interval)).setCount(String.valueOf(entry.getDocCount())));
        }
        return DimAnalysisRsp.R.of().setInfo(data).setDimName(cacheInfo.getDimName()).setClassify(cacheInfo.getClassify()).setDimId(cacheInfo.getDimId());
    }

    private SearchRequest getDateHistogramRequest(String dimIdColumn, QueryBuilder queryBuilder, DateHistogramInterval interval,String indexId) {

        IndexNameInfo indexNameInfo = CacheUtils.getCacheInfo(CacheConfig.CONFIGURATION_INFO, indexId);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(queryBuilder).trackTotalHits(true).size(0);
        DateHistogramAggregationBuilder builder =
                AggregationBuilders.dateHistogram("count").field(dimIdColumn).dateHistogramInterval(interval).format(
                        "yyyy-MM-dd").order(BucketOrder.key(true)).minDocCount(1);
        searchSourceBuilder.aggregation(builder);
        return new SearchRequest(indexNameInfo.getDataDate()).source(searchSourceBuilder);
    }

    private DimAnalysisRsp.R getDateResult(String column, QueryBuilder queryBuilder, DimInfo cacheInfo,
                                         List<GroupItem> items,String indexId) throws IOException {
        SearchRequest request = getRequest(column, queryBuilder, cacheInfo.getClassify(), items,indexId);
        return getResult(ESQueryUtils.getClient().search(request, RequestOptions.DEFAULT), cacheInfo);
    }

    /**
     * 描述和标志标签
     * @param model
     * @return
     */
    private DimAnalysisRsp.R getStringResult(CustomersAnalysisModel model, DimInfo cacheInfo) throws IOException {
        // 筛选条件
        QueryBuilder queryBuilder = EsQueryBuilder.buildFilter(model.getFilterGroup());
        Map<String, String> relation = dateService.symbolRelation(model.getIndexId());
        String column = FieldHandler.addSuffixField(model.getField(), model.getFieldsDataDate(),relation);
        TermsAggregationBuilder termsBuilder =
                AggregationBuilders.terms("count").field(column).order(BucketOrder.key(true)).size(Integer.MAX_VALUE);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.aggregation(termsBuilder).query(queryBuilder).trackTotalHits(true).size(0);

        IndexNameInfo indexNameInfo = CacheUtils.getCacheInfo(CacheConfig.CONFIGURATION_INFO, model.getIndexId());
        SearchRequest searchRequest = new SearchRequest(indexNameInfo.getDataDate()).source(searchSourceBuilder);
        SearchResponse response = ESQueryUtils.getClient().search(searchRequest, RequestOptions.DEFAULT);
        Aggregations aggregations =  response.getAggregations();

        Terms terms = aggregations.get("count");
        List<DimAnalysisRsp.Info> list = new ArrayList<>();
        for(Terms.Bucket entry : terms.getBuckets()){
            String title = Optional
                    .ofNullable((DimItemInfo) CacheUtils.getCacheInfo(CacheConfig.DIM_ITEM_INFO,
                            String.format(BaseParse.GROUP_FORMAT, model.getField(), entry.getKeyAsString())))
                    .map(DimItemInfo::getDimItemName)
                    .orElse(entry.getKeyAsString());
            if ("06".equals(cacheInfo.getClassify())){
                title = joinStr(title,entry.getKeyAsString());
            }
            list.add(DimAnalysisRsp.Info.of().setTitle(title).setCount(String.valueOf(entry.getDocCount())));
        }
        return DimAnalysisRsp.R.of().setDimName(cacheInfo.getDimName()).setInfo(list).setClassify(cacheInfo.getClassify()).setDimId(cacheInfo.getDimId());
    }

    public String joinStr(String key,String value){
        StringBuffer buffer = new StringBuffer();
        buffer.append(key).append("(").append(value).append(")");
        return buffer.toString();
    }
}
